import java.io.IOException;
import java.util.*;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

/**
 * Reducer class for the distributed Breadth First Search.  Essentially this
 * performs a min on the distances to the given key and emmits <key,
 * Dist:link0;link1;...;link2> Determines the list of links from the first
 * pair emmitted by a map job - the key value pair containing the old
 * distance.
 *
 * @author Ryan Hofler
 * @version 0.1
 */
public class DBFSReducer extends MapReduceBase
    implements Reducer<Text, Text, Text, Text> {
    public void reduce(Text key,
		       Iterator<Text> values,
		       OutputCollector<Text, Text> output,
		       Reporter reporter)
	throws IOException {

        int minDist = Integer.MAX_VALUE;
        
        String link_list = "";
        
        while(values.hasNext()){
            // Split the value into distance (first index) and links (second)
            String[] dist_links = values.next().toString().split("\\t");
            
            // only one should have links
            if(dist_links.length > 1)
                link_list = dist_links[1];
            
            // get the best dist so far
            int dist = Integer.parseInt(dist_links[0]);            
            minDist = Math.min(minDist, dist);
        }

        // Collect the key with best dist and links
        output.collect(key, new Text(minDist+"\t"+link_list));
    }
}

